#include <functional>
#include <fermat/net/tcp_server.h>
#include <fermat/common/logging.h>
#include <fermat/net/tcp_acceptor.h>
#include <fermat/net/socket_option.h>
#include <fermat/event/event_loop.h>
#include <fermat/event/event_loop_thread_pool.h>
#include <fermat/net/tcp_socket.h>
#include <fermat/common/format.h>

namespace fermat {
namespace net {

TcpServer::TcpServer(fermat::event::EventLoopPtr loop,
                     const SocketAddress& listenAddr,
                     const std::string& nameArg,
                     Option option)
  : _loop(loop), 
    _hostport(listenAddr.toStringHostPort()),
    _name(nameArg),
    _acceptor(new Acceptor(loop, listenAddr, option == kReusePort)),
    _threadPool(new fermat::event::EventLoopThreadPool(loop, nameArg)),
    _connectionCallback(defaultConnectionCallback),
    _messageCallback(defaultMessageCallback),
    _nextConnId(1)
{
  _acceptor->setNewConnectionCallback(
      std::bind(&TcpServer::newConnection, this, std::placeholders::_1, std::placeholders::_2));
}

TcpServer::~TcpServer()
{
    _loop->assertInLoopThread();

    for (ConnectionMap::iterator it(_connections.begin());
        it != _connections.end(); ++it) {
        TcpConnectionPtr conn = it->second;
        it->second.reset();
        conn->getLoop()->runInLoop(
          std::bind(&TcpConnection::connectDestroyed, conn));
        conn.reset();
    }
}

void TcpServer::setThreadNum(int numThreads)
{
  assert(0 <= numThreads);
  _threadPool->setThreadNum(numThreads);
}

void TcpServer::start()
{
    if (_started.fetch_increment() == 0) {
        _threadPool->start(_threadInitCallback);
        assert(!_acceptor->listenning());
        
        _loop->runInLoop(
            std::bind(&Acceptor::listen, _acceptor.get()));
    }
}

void TcpServer::newConnection(int sockfd, const SocketAddress& peerAddr)
{
    _loop->assertInLoopThread();
    fermat::event::EventLoopPtr ioLoop = _threadPool->getNextLoop();
    fermat::MemoryWriter w;
    w.write("{}:{}#{}", _name, _hostport, _nextConnId);
    ++_nextConnId;
    std::string connName = w.str();
    SocketAddress localAddr = SocketOption::getLocalAddress(sockfd);

    TcpConnectionPtr conn(new TcpConnection(ioLoop,
                                            connName,
                                            sockfd,
                                            localAddr,
                                            peerAddr));
    _connections[connName] = conn;
    conn->setConnectionCallback(_connectionCallback);
    conn->setMessageCallback(_messageCallback);
    conn->setWriteCompleteCallback(_writeCompleteCallback);
    conn->setCloseCallback(
        std::bind(&TcpServer::removeConnection, this, std::placeholders::_1)); 
    ioLoop->runInLoop(std::bind(&TcpConnection::connectEstablished, conn));
}

void TcpServer::removeConnection(const TcpConnectionPtr& conn)
{

  _loop->runInLoop(std::bind(&TcpServer::removeConnectionInLoop, this, conn)
);
}

void TcpServer::removeConnectionInLoop(const TcpConnectionPtr& conn)
{
  _loop->assertInLoopThread();
  size_t n = _connections.erase(conn->name());
  (void)n;
  assert(n == 1);
  EventLoopPtr ioLoop = conn->getLoop();
  ioLoop->queueInLoop(
      std::bind(&TcpConnection::connectDestroyed, conn));
}
}
}
